package com.hmdp.schedule;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.hmdp.domain.entity.BlogLike;
import com.hmdp.mapper.BlogLikeMapper;
import com.hmdp.service.BlogLikeService;
import io.lettuce.core.RedisCommandExecutionException;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.sql.SQLException;
import java.time.Duration;
import java.util.*;

import static com.hmdp.utils.RedisConstants.*;

@Component
@EnableScheduling
@Slf4j
public class RedisLikeBlobSchedule {
    @Autowired
    BlogLikeMapper blogLikeMapper;
    @Autowired
    StringRedisTemplate stringRedisTemplate;
    @Autowired
    BlogLikeService blogLikeService;
    @Resource(name = "likeListData")
    DefaultRedisScript<List> likeListData = new DefaultRedisScript<>();

    @Scheduled(fixedDelay = 60 * 60 * 72 * 1000, initialDelay = 60 * 1000)
    @Async("scheduleThreadPool")
    public void flushDb() {
        log.info("点赞数据准备同步给数据库");
        ArrayList<ArrayList<String>> dataList = (ArrayList<ArrayList<String>>) stringRedisTemplate.execute(likeListData, Collections.emptyList(), new String());
        if (CollUtil.isEmpty(dataList)) {
            return;
        }
        if (dataList.get(0).size() == 0 && dataList.get(1).size() == 0) {
            return;
        }
        //放入stream流中
        innerStream(dataList);
        //开启缓存同步线程
        cacheHandler();
    }

    private void innerStream(ArrayList<ArrayList<String>> dataList) {
        RedisCacheStream cacheStream = new RedisCacheStream();
        cacheStream.setDelete(dataList.get(1));
        cacheStream.setSave(dataList.get(0));
        ObjectRecord<String, RedisCacheStream> record = ObjectRecord.create(BLOG_LIKED_CACHE, cacheStream);
        stringRedisTemplate.opsForStream().add(record);
    }

    @PostConstruct
    public void init() {
        try {
            if (Boolean.FALSE.equals(stringRedisTemplate.hasKey(BLOG_LIKED_CACHE))) {
                //如果当前stream流未创建出来，是不能创建消费者组的
                RecordId recordId = initStream();
                createGroup();
                stringRedisTemplate.opsForStream().delete(BLOG_LIKED_CACHE, recordId);
                return;
            }
            createGroup();
        } catch (RedisCommandExecutionException e) {
            e.printStackTrace();
            log.error("创建消费者组失败:{}", e.getMessage());
        }


    }

    private RecordId initStream() {
        RedisCacheStream cacheStream = new RedisCacheStream();
        cacheStream.setDelete(new ArrayList<>());
        cacheStream.setSave(new ArrayList<>());
        ObjectRecord<String, RedisCacheStream> record = ObjectRecord.create(BLOG_LIKED_CACHE, cacheStream);
        return stringRedisTemplate.opsForStream().add(record);
    }

    private void createGroup() {
        StreamInfo.XInfoGroups groups = stringRedisTemplate.opsForStream().groups(BLOG_LIKED_CACHE);
        if (groups.isEmpty()) {
            //消费历史记录流中所有的数据
            stringRedisTemplate.opsForStream().createGroup(BLOG_LIKED_CACHE, ReadOffset.from("0"), BLOG_LIKED_CACHE_GROUP);
        }
    }

    public void cacheHandler() {
        if (Boolean.FALSE.equals(stringRedisTemplate.hasKey(BLOG_LIKED_CACHE))) {
            createGroup();
        }else {
            return;
        }
        while (true) {
            List<ObjectRecord<String, RedisCacheStream>> recordList = null;
            try {
                recordList = stringRedisTemplate.opsForStream().read(RedisCacheStream.class,
                        Consumer.from(BLOG_LIKED_CACHE_GROUP, BLOG_LIKED_CACHE_GROUP_CONSUMER),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                        StreamOffset.create(BLOG_LIKED_CACHE, ReadOffset.from(">")));
                if (CollUtil.isEmpty(recordList)) {
                    continue;
                }
                ObjectRecord<String, RedisCacheStream> consumer = consumer(recordList);
                stringRedisTemplate.opsForStream().acknowledge(BLOG_LIKED_CACHE_GROUP, consumer);
            } catch (Exception e) {
                log.error("点赞数据同步到数据库失败:{}", e.getMessage());
                //删除重复数据
                if (e.getMessage().contains("UK_praise")) {
                    deleteRepeateData(recordList);
                } else {
                    e.printStackTrace();
                }
                try {
                    handerException();
                } catch (Exception exception) {
                    exception.printStackTrace();
                }
            }
        }
    }

    private void deleteRepeateData(List<ObjectRecord<String, RedisCacheStream>> recordList) {
        assert recordList != null;
        ObjectRecord<String, RedisCacheStream> objectRecord = recordList.get(0);
        RedisCacheStream cacheStream = objectRecord.getValue();
        //获取重复数据
        Set<String> intersect = getRepeateData(cacheStream);
        if (CollUtil.isEmpty(intersect)) {
            //如果说交集为空的话，说明这个消息不对
            stringRedisTemplate.opsForStream().delete(objectRecord);
            //将错误的消息存放到日志中，交给人工检查
            log.error("消息错误:{}", objectRecord);
        }
        //删除重复数据
        deleteRepeateData(cacheStream, intersect);
        log.warn("有重复数据，已删除:{}", intersect);
        //更换stream中的消息
        stringRedisTemplate.opsForStream().acknowledge(BLOG_LIKED_CACHE, BLOG_LIKED_CACHE_GROUP, objectRecord.getId());
        //删除数据后，集合不为空才能放入stream
        if (!CollUtil.isEmpty(cacheStream.getSaveBlogLike()) || !CollUtil.isEmpty(cacheStream.getDeleteBlogLike())) {
            ObjectRecord<String, RedisCacheStream> newRecord = ObjectRecord.create(BLOG_LIKED_CACHE, cacheStream);
            stringRedisTemplate.opsForStream().add(newRecord);
        }
    }

    private static void deleteRepeateData(RedisCacheStream cacheStream, Set<String> intersect) {
        cacheStream.jsonToBean();
        intersect.forEach(s -> {
            BlogLike blogLike = JSONUtil.toBean(s, BlogLike.class);
            cacheStream.getSaveBlogLike().remove(blogLike);
        });
        cacheStream.beanToJson();
    }

    /**
     * 查询数据库，与redis的set集合求交集
     * 获取重复数据
     *
     * @param cacheStream redis中的数据
     * @return
     */
    private Set<String> getRepeateData(RedisCacheStream cacheStream) {
        cacheStream.jsonToBean();
        ArrayList<Integer> ids = new ArrayList<>();
        cacheStream.getSaveBlogLike().forEach(blogLike -> {
            stringRedisTemplate.opsForSet().add(BLOG_LIKED_CACHE_TEMPORARY_STORAGE + ":redis", JSONUtil.toJsonStr(blogLike));
            ids.add(blogLike.getBlogId());
        });
        //查数据库
        QueryWrapper<BlogLike> blogLikeQueryWrapper = new QueryWrapper<>();
        blogLikeQueryWrapper.in("blog_id", ids);
        List<BlogLike> blogLikes = blogLikeMapper.selectList(blogLikeQueryWrapper);
        blogLikes.forEach(blogLike -> {
            stringRedisTemplate.opsForSet().add(BLOG_LIKED_CACHE_TEMPORARY_STORAGE + ":MySQL", JSONUtil.toJsonStr(blogLike));
        });
        ArrayList<String> keys = new ArrayList<>();
        keys.add(BLOG_LIKED_CACHE_TEMPORARY_STORAGE + ":redis");
        keys.add(BLOG_LIKED_CACHE_TEMPORARY_STORAGE + ":MySQL");
        Set<String> intersect = stringRedisTemplate.opsForSet().intersect(keys);
        stringRedisTemplate.delete(keys);
        return intersect;
    }

    /**
     * 同步数据到Mysql
     *
     * @param recordList
     */
    @Transactional(rollbackFor = SQLException.class)
    public ObjectRecord<String, RedisCacheStream> consumer(List<ObjectRecord<String, RedisCacheStream>> recordList) {
        ObjectRecord<String, RedisCacheStream> record = recordList.get(0);
        RedisCacheStream redisCacheStream = record.getValue();
        redisCacheStream.jsonToBean();
        ArrayList<BlogLike> deleteBlogLike = redisCacheStream.getDeleteBlogLike();
        if (deleteBlogLike != null) {
                blogLikeMapper.deleteBatchByUserIdAndBlogId(deleteBlogLike);
        }
        if (redisCacheStream.getSaveBlogLike() != null) {
            blogLikeService.saveBatch(redisCacheStream.getSaveBlogLike());
        }
        log.info("点赞缓存已同步!");
        return record;
    }

    public void handerException() {
        while (true) {
            List<ObjectRecord<String, RedisCacheStream>> recordList = stringRedisTemplate.opsForStream().read(
                    RedisCacheStream.class,
                    Consumer.from(BLOG_LIKED_CACHE_GROUP, BLOG_LIKED_CACHE_GROUP_CONSUMER),
                    StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                    StreamOffset.create(BLOG_LIKED_CACHE, ReadOffset.from("0")));
            if (recordList == null || recordList.isEmpty() || recordList.get(0) == null) {
                break;
            }
            ObjectRecord<String, RedisCacheStream> record = recordList.get(0);
            //死信处理
            stringRedisTemplate.opsForStream().pending(
                    Objects.requireNonNull(record.getStream()),
                    BLOG_LIKED_CACHE_GROUP,
                    Range.unbounded(),
                    10).forEach(pendingMessage -> {
                stringRedisTemplate.opsForStream().acknowledge(BLOG_LIKED_CACHE, record);
            });

            //同步数据
            ObjectRecord<String, RedisCacheStream> consumer = consumer(recordList);
            stringRedisTemplate.opsForStream().acknowledge(BLOG_LIKED_CACHE_GROUP, consumer);
        }
    }

    @Data
    static class RedisCacheStream {
        private ArrayList<String> delete;
        private ArrayList<String> save;
        private ArrayList<BlogLike> deleteBlogLike;
        private Set<BlogLike> saveBlogLike;

        public void jsonToBean() {
            //保证只初始化一次
            if (!CollUtil.isEmpty(delete) && CollUtil.isEmpty(deleteBlogLike)) {
                for (String json : delete) {
                    deleteBlogLike = new ArrayList<>();
                    deleteBlogLike.add(JSONUtil.toBean(json, BlogLike.class));
                }
            }
            if (!CollUtil.isEmpty(save) && CollUtil.isEmpty(saveBlogLike)) {
                saveBlogLike = new LinkedHashSet<>();
                for (String json : save) {
                    saveBlogLike.add(JSONUtil.toBean(json, BlogLike.class));
                }
            }
        }

        /**
         * 如果saveBlogLike有值，重构save集合
         */
        public void beanToJson() {
            if (CollUtil.isEmpty(saveBlogLike)) {
                save = new ArrayList<>();
                saveBlogLike.forEach(blogLike -> {
                    save.add(JSONUtil.toJsonStr(blogLike));
                });
            }
        }

    }

}
